package com.adjaisd.scarecrow.common.utils.mqtt;

import com.adjaisd.scarecrow.common.config.MqttConfig;
import com.adjaisd.scarecrow.dto.DeviceInfoDTO;
import com.adjaisd.scarecrow.entity.DeviceGps;
import com.adjaisd.scarecrow.service.*;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.*;

@Slf4j
@Component
public class EmqClient implements MqttCallback {

    @Autowired
    private MqttConfig mqttConfig;  // 配置类

    @Autowired
    private RedisService redisService;

    private MqttClient mqttClient;  // 客户端连接

    public void connect() {
        Map<String, String> map = MqttUtils.getToken(mqttConfig, redisService);
        String username = map.get("username");
        String password = map.get("password");
        String clientId = "monitor." + UUID.randomUUID();
        mqttClient = MqttUtils.connect(
                mqttConfig.getBrokerUrl(), clientId,
                username, password, this);
    }

    @Async("asyncServiceExecutor")
    public void publish(String topic, String msg, int qos) {
        MqttUtils.publish(mqttClient, topic, msg, qos);
    }

    @Async("asyncServiceExecutor")
    public void subscribe(String topicName) throws MqttException {
        MqttUtils.subscribe(mqttClient, topicName);
        System.out.println("订阅成功 : " + topicName);
    }

    @Async("asyncServiceExecutor")
    public void unsubscribe(String topicName) throws MqttException {
        MqttUtils.unsubscribe(mqttClient, topicName);
        System.out.println("取消订阅成功 : " + topicName);
    }

    // -------------------------

    @Autowired
    QuotaService quotaService;

    @Override
    public void connectionLost(Throwable throwable) {
        connect();
        try {
            for (String subject : quotaService.getAllSubject())
                subscribe("$queue/" + subject);
            subscribe("$queue/" + "device/gps");
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Autowired
    AlarmService alarmService;

    @Autowired
    DeviceStatusService deviceStatusService;

    @Autowired
    DeviceGpsService deviceGpsService;

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws JsonProcessingException {
        //接收到消息
        String payload = new String(mqttMessage.getPayload());
        ObjectMapper mapper = new ObjectMapper();
        Map payloadMap = mapper.readValue(payload, Map.class);

        System.out.println("从主题：" + topic + "\t接收到数据：" + payloadMap);

        //解析指标
        if (!topic.equals("device/gps")) {
            DeviceInfoDTO deviceInfoDTO = quotaService.analysis(topic, payloadMap);
            if (deviceInfoDTO != null) {
                //告警判断
                deviceInfoDTO = alarmService.verifyDeviceInfo(deviceInfoDTO);
                //保存设备信息
                deviceStatusService.saveDeviceInfo(deviceInfoDTO.getDeviceStatusDTO());
                //保存指标数据
                quotaService.saveQuotaToInflux(deviceInfoDTO.getQuotaDTOList());
                //指标透传
                // noticeService.quotaTransfer(deviceInfoDTO.getQuotaList());
            }
        } else {
            //解析gps
            DeviceGps deviceGps = deviceGpsService.analysis(topic, payloadMap);
            if (deviceGps != null) {
                System.out.println("gps解析结果：" + deviceGps);
                // 更新GPS到mysql与influx
                deviceGpsService.updateDeviceGPS(deviceGps);
                // noticeService.gpsTransfer(deviceLocation);
            }
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

    }

}
